/*-------------------------------------------------------------------------
 *
 * async.c
 *	  异步通知：NOTIFY, LISTEN, UNLISTEN
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/commands/async.c
 *
 *-------------------------------------------------------------------------
 */

/*-------------------------------------------------------------------------
 * 异步通知模型截至9.0版本：
 *
 * 1. 同一机器上的多个后端。多个后端监听
 *	  几个通道。（在代码的其他部分，通道也称为“条件”。）
 *
 * 2. 在磁盘存储中有一个中央队列（目录 pg_notify/），
 *	  由 slru.c 模块将活动使用的页面映射到共享内存中。
 *	  所有通知消息都放置在队列中，稍后由监听后端读取。
 *
 *	  没有中央知识知道哪个后端监听哪个通道；
 *	  每个后端都有自己感兴趣的通道列表。
 *
 *	  尽管只有一个队列，但通知被视为数据库本地；
 *	  这是通过在每个通知消息中包含发送者的数据库 OID
 *	  来实现的。监听后端忽略与其数据库 OID 不匹配的消息。
 *	  这很重要，因为它确保发送者和接收者具有相同的数据库编码，
 *	  不会误解通道名称或有效负载字符串中的非 ASCII 文本。
 *
 *	  由于通知预计不会在数据库崩溃后保留，
 *	  我们可以在每次重启时简单清理 pg_notify 数据，
 *	  无需 WAL 支持或 fsync。
 *
 * 3. 每个至少在一个通道上监听的后端通过
 *	  将其 PID 录入 AsyncQueueControl 的数组进行注册。然后它扫描
 *	  中央队列中的所有传入通知，首先将通知的数据库 OID
 *	  与它自己的数据库 OID 进行比较，然后将被通知的通道
 *	  与它正在监听的通道列表进行比较。如果有匹配，
 *	  则将通知事件传递给其前端。未匹配的事件会被简单跳过。
 *
 * 4. NOTIFY 语句（例程 Async_Notify）将通知存储在
 *	  后端本地列表中，直到事务结束后才会处理。
 *
 *	  相同事务中的重复通知仅作为一个通知发送。
 *	  这是为了减少工作量，例如，当一个触发器在一个有两百万行的表上
 *	  为每一行发送通知时。如果应用程序需要接收每一个已发送的通知，
 *	  它可以轻松地在额外的有效负载参数中添加一些唯一字符串。
 *
 *	  当事务准备提交时，PreCommit_Notify() 将待处理的通知
 *	  添加到队列的头部。队列的头指针始终指向下一个空闲位置，
 *	  位置只是一个页面编号和该页面中的偏移量。在将事务标记为已提交之前，执行此操作。
 *	  如果在写入通知时遇到问题，我们仍然可以调用 elog(ERROR, ...)，
 *	  事务将会回滚。
 *
 *	  一旦我们将所有通知放入队列中，我们将返回
 *	  CommitTransaction()，然后执行实际的事务提交。
 *
 *	  提交后，我们将再次被调用（AtCommit_Notify()）。
 *	  这里我们对有效的监听状态（listenChannels）进行任何实际更新。
 *	  然后我们向可能对我们的消息感兴趣的任何后端发信号
 *	  （包括我们的后端，如果正在监听）。这是通过
 *	  SignalBackends() 实现的，它扫描监听后端列表
 *	  并向每个监听后端发送 PROCSIG_NOTIFY_INTERRUPT 信号
 *	  （我们不知道哪个后端正在监听哪个通道，因此必须通知所有后端）。
 *	  不过，我们可以排除那些已经最新的后端，
 *	  我们也可以排除那些在其他数据库中的后端（除非它们落后太多，
 *	  应该被踢出来以使它们推进其指针）。
 *
 *	  最后，当我们完全退出事务并即将闲置时，
 *	  我们扫描队列中需要发送到前端的消息
 *	  （可能是来自其他后端的通知，或来自我们自己后端的自我通知）。
 *	  这一环节不是 CommitTransaction 序列的一部分，
 *	  原因有两个。首先，在将数据发送到我们的前端时可能会出现错误，
 *	  在提交后清理时发生错误是非常糟糕的。其次，在一个过程
 *	  在单个前端命令内发出多次提交的情况下，我们不希望在命令完成之前
 *	  将通知发送到我们的前端；但是对其他后端的通知应在每次提交后
 *	  立即发送。
 *
 * 5. 在接收到 PROCSIG_NOTIFY_INTERRUPT 信号时，
 *	  信号处理程序设置进程的锁存器，
 *	  这会立即触发事件进行处理，如果该后端处于空闲状态
 *	  （即，它正在等待前端命令，并且不在事务块内。参见
 *	  ProcessClientReadInterrupt()）。否则，处理程序可能只会设置一个
 *	  标志，这会导致处理在我们下次进入空闲状态之前发生。
 *
 *	  入站通知处理包括读取自上次扫描以来到达的所有通知。
 *	  我们读取每个通知，直到我们到达未提交事务的通知或
 *	  头指针的位置为止。
 *
 * 6. 为了避免 SLRU 循环和限制磁盘空间消耗，
 *	  需要推进尾指针，以便可以截断旧页面。
 *	  这相对昂贵（特别是，它需要独占锁），
 *	  因此我们不希望过于频繁地进行此操作。我们让发送后端
 *	  进行此工作，如果它们将队列头推进到了一个新页面，
 *	  但每 QUEUE_CLEANUP_DELAY 页面只执行一次。
 *
 * 监听同一通道的应用程序将收到
 * 自己的 NOTIFY 的 NOTIFY 消息。如果没有用，可以通过将
 * NOTIFY 消息中的 be_pid 与应用程序自身后端的 PID 进行比较来忽略这些消息。
 * （截至 FE/BE 协议 2.0，后端的 PID 在启动时提供给前端。）
 * 上述设计确保不会因忽略自我通知而错过来自其他后端的通知。
 *
 * 用于通知管理的共享内存量（NUM_NOTIFY_BUFFERS）
 * 可以在不影响任何内容的情况下调整，只会影响性能。
 * 能够一次排队的最大通知数据量由 slru.c 的循环限制决定；
 * 见下文 QUEUE_MAX_PAGE。
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include <limits.h>
#include <unistd.h>
#include <signal.h>

#include "access/parallel.h"
#include "access/slru.h"
#include "access/transam.h"
#include "access/xact.h"
#include "catalog/pg_database.h"
#include "commands/async.h"
#include "common/hashfn.h"
#include "funcapi.h"
#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "miscadmin.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/procsignal.h"
#include "storage/sinval.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/ps_status.h"
#include "utils/snapmgr.h"
#include "utils/timestamp.h"


/*
 * NOTIFY 有效负载的最大大小，包括终止的 NULL。这个必须足够小，以便通知消息适合一个 SLRU 页面。这里的魔法修正因子是非关键的，只要它大于 AsyncQueueEntryEmptySize --- 我们将其设置得明显大于这个值，因此该数据结构的变化不会影响用户可见的限制。
 */
#define NOTIFY_PAYLOAD_MAX_LENGTH	(BLCKSZ - NAMEDATALEN - 128)

/*
 * 表示全局通知队列中一个条目的结构
 *
 * 这个结构的声明具有最大的长度，但在实际的队列条目中，数据区域仅足够容纳实际的通道和有效负载字符串（每个以 NULL 终止）。AsyncQueueEntryEmptySize 是可能的最小条目大小，如果通道和有效负载字符串都为空（但请注意，它不包括对齐填充）。
 *
 * “长度”字段应始终四舍五入到下一个 QUEUEALIGN 的倍数，以确保所有字段都正确对齐。
 */
typedef struct AsyncQueueEntry
{
	int			length;			/* 条目总分配长度 */
	Oid			dboid;			/* 发送者的数据库 OID */
	TransactionId xid;			/* 发送者的 XID */
	int32		srcPid;			/* 发送者的 PID */
	char		data[NAMEDATALEN + NOTIFY_PAYLOAD_MAX_LENGTH];
} AsyncQueueEntry;

/* 当前，AsyncQueueEntry 的任何字段都不需要超过 int 对齐 */
#define QUEUEALIGN(len)		INTALIGN(len)

#define AsyncQueueEntryEmptySize	(offsetof(AsyncQueueEntry, data) + 2)

/*
 * 描述队列位置的结构，以及处理它的各种宏
 */
typedef struct QueuePosition
{
	int			page;			/* SLRU 页面编号 */
	int			offset;			/* 页面内的字节偏移 */
} QueuePosition;

#define QUEUE_POS_PAGE(x)		((x).page)
#define QUEUE_POS_OFFSET(x)		((x).offset)

#define SET_QUEUE_POS(x,y,z) \
	do { \
		(x).page = (y); \
		(x).offset = (z); \
	} while (0)

#define QUEUE_POS_EQUAL(x,y) \
	((x).page == (y).page && (x).offset == (y).offset)

#define QUEUE_POS_IS_ZERO(x) \
	((x).page == 0 && (x).offset == 0)

/* 选择逻辑上较小的 QueuePosition */
#define QUEUE_POS_MIN(x,y) \
	(fc_asyncQueuePagePrecedes((x).page, (y).page) ? (x) : \
	 (x).page != (y).page ? (y) : \
	 (x).offset < (y).offset ? (x) : (y))

/* 选择逻辑上较大的 QueuePosition */
#define QUEUE_POS_MAX(x,y) \
	(fc_asyncQueuePagePrecedes((x).page, (y).page) ? (y) : \
	 (x).page != (y).page ? (x) : \
	 (x).offset > (y).offset ? (x) : (y))

/*
 * 确定我们多频繁尝试推进尾指针的参数：
 * 我们在每 QUEUE_CLEANUP_DELAY 页的 NOTIFY 数据之后执行此操作。这也是在我们决定需要唤醒它以推进其指针之前，另一个数据库中的后端需要落后多少的距离。
 *
 * 抵制将这个值设置得过大的诱惑。虽然这在某些地方可以节省工作，但在其他地方会增加成本。特别是，这个值应该小于 NUM_NOTIFY_BUFFERS，以确保后端赶上需要读取的页面在 SLRU 缓存中未过期。
 */
#define QUEUE_CLEANUP_DELAY 4

/*
 * 描述正在侦听的后端状态的结构
 */
typedef struct QueueBackendStatus
{
	int32		pid;			/* 要么是 PID 要么是 InvalidPid */
	Oid			dboid;			/* 后端的数据库 OID，或 InvalidOid */
	BackendId	nextListener;	/* 下一个侦听器的 ID，或 InvalidBackendId */
	QueuePosition pos;			/* 后端已读取队列到此 */
} QueueBackendStatus;

/*
 * 共享内存状态用于 LISTEN/NOTIFY（不包括其 SLRU 相关内容）
 *
 * AsyncQueueControl 结构由 NotifyQueueLock 和 NotifyQueueTailLock 保护。
 *
 * 当以共享模式持有 NotifyQueueLock 时，后端只能检查自己的条目以及头指针和尾指针。因此，我们可以允许后端在仅持有共享锁的情况下更新自己的记录（因为没有其他后端会检查它）。
 *
 * 当以独占模式持有 NotifyQueueLock 时，后端可以检查其他后端的条目并且还可以更改头指针。当同时以独占模式持有 NotifyQueueLock 和 NotifyQueueTailLock 时，后端可以更改尾指针。
 *
 * NotifySLRULock 用作 pg_notify SLRU 缓冲区的控制锁。为了避免死锁，每当我们需要多个锁时，我们首先获取 NotifyQueueTailLock，然后是 NotifyQueueLock，最后是 NotifySLRULock。
 *
 * 每个后端使用 backend[] 数组中索引等于其 BackendId 的条目（范围从 1 到 MaxBackends）。我们依赖这一点使 SendProcSignal 快速。
 *
 * 积极监听的后端的 backend[] 数组条目通过 firstListener 和 nextListener 链接串联在一起，以便我们可以扫描它们，而不必遍历非活动条目。我们按 BackendId 保持此列表的顺序，以便在有多个活动条目时扫描是缓存友好的。
 */
typedef struct AsyncQueueControl
{
	QueuePosition head;			/* 头指针指向下一个空闲位置 */
	QueuePosition tail;			/* 尾指针必须 <= 每个
								 * 正在监听的后端的队列位置 */
	int			stopPage;		/* 最旧的未回收页面；必须 <=
								 * tail.page */
	BackendId	firstListener;	/* 第一个监听者的 id，或 InvalidBackendId */
	TimestampTz lastQueueFillWarn;	/* 最后一次队列满消息的时间 */
	QueueBackendStatus backend[FLEXIBLE_ARRAY_MEMBER];
	/* backend[0] 不被使用；已使用的条目从 [1] 到 [MaxBackends] */
} AsyncQueueControl;

static AsyncQueueControl *asyncQueueControl;

#define QUEUE_HEAD					(asyncQueueControl->head)
#define QUEUE_TAIL					(asyncQueueControl->tail)
#define QUEUE_STOP_PAGE				(asyncQueueControl->stopPage)
#define QUEUE_FIRST_LISTENER		(asyncQueueControl->firstListener)
#define QUEUE_BACKEND_PID(i)		(asyncQueueControl->backend[i].pid)
#define QUEUE_BACKEND_DBOID(i)		(asyncQueueControl->backend[i].dboid)
#define QUEUE_NEXT_LISTENER(i)		(asyncQueueControl->backend[i].nextListener)
#define QUEUE_BACKEND_POS(i)		(asyncQueueControl->backend[i].pos)

/*
 * 通过 SLRU 缓冲区访问通知队列的区域
 */
static SlruCtlData NotifyCtlData;

#define NotifyCtl					(&NotifyCtlData)
#define QUEUE_PAGESIZE				BLCKSZ
#define QUEUE_FULL_WARN_INTERVAL	5000	/* 最多每 5 秒警告一次 */

/*
 * 使用段 0000 到 FFFF。每个段包含 SLRU_PAGES_PER_SEGMENT 页，
 * 这使我们获得从 0 到 SLRU_PAGES_PER_SEGMENT * 0x10000 - 1 的页面。
 * 我们可以使用 SlruScanDirectory() 允许的任意数量的段，但这已经给我们提供了这么多空间，似乎不值得麻烦。
 *
 * 我们一次可以在队列中拥有的最大数据量是 QUEUE_MAX_PAGE/2 页，
 * 因为更多的话会让 slru.c 认为发生了环绕条件。使用默认的 BLCKSZ，这意味着队列中可以有多达 8GB 的待处理数据。
 *
 * 注意：可以将 QUEUE_MAX_PAGE 重新定义为 SLRU_PAGES_PER_SEGMENT 的较小倍数，以便于测试队列满的行为。
 */
#define QUEUE_MAX_PAGE			(SLRU_PAGES_PER_SEGMENT * 0x10000 - 1)

/*
 * listenChannels 确定我们实际监听的通道
 * （即，已提交 LISTEN 的通道）。它是一个简单的通道名称列表，
 * 分配在 TopMemoryContext 中。
 */
static List *listenChannels = NIL;	/* C 字符串列表 */

/*
 * 挂起的 LISTEN/UNLISTEN 操作的状态由当前事务中请求的所有操作的有序列表组成。正如上面所解释的，我们实际上不会在事务提交之前更改 listenChannels。
 *
 * 该列表保存在 CurTransactionContext 中。在子事务中，每个子事务在其自己的 CurTransactionContext 中都有自己的列表，但成功的子事务将其列表附加到其父列表上。失败的子事务则简单地丢弃它们的列表。
 */
typedef enum
{
	LISTEN_LISTEN,
	LISTEN_UNLISTEN,
	LISTEN_UNLISTEN_ALL
} ListenActionKind;

typedef struct
{
	ListenActionKind action;
	char		channel[FLEXIBLE_ARRAY_MEMBER]; /* 以 null 结尾的字符串 */
} ListenAction;

typedef struct ActionList
{
	int			nestingLevel;	/* 当前事务嵌套深度 */
	List	   *actions;		/* ListenAction 结构的列表 */
	struct ActionList *upper;	/* 上层事务级别的详细信息 */
} ActionList;

static ActionList *pendingActions = NULL;

/*
 * 出站通知的状态由当前事务中所有已 NOTIFY 的通道 + 有效载荷的列表组成。我们实际上不会执行 NOTIFY，直到事务提交。 如果当前（子）事务中没有执行 ANY NOTIFY，则 pendingNotifies 为 NULL。
 *
 * 我们会丢弃在同一事务中发出的重复通知事件。因此，除了我们需要跟踪事件顺序的列表（因为我们保证按顺序交付它们）之外，我们还构建一个哈希表，以便检测重复事件。因为构建哈希表的成本相对较高，所以我们只在当前（子）事务中至少排队有 MIN_HASHABLE_NOTIFIES 个事件时才这样做；在此之前，我们只是线性扫描事件。
 *
 * 该列表保存在 CurTransactionContext 中。在子事务中，每个子事务在其自己的 CurTransactionContext 中都有自己的列表，但成功的子事务会将其条目添加到父列表中。失败的子事务则简单地丢弃它们的列表。由于这些列表是独立的，因此子事务的列表中可能有与某些祖先（子）事务中的事件重复的通知事件；在将子事务的列表合并到其父列表时，我们会去掉重复项。
 *
 * 注意：在事务中，操作和通知列表不会相互作用。特别是，如果一个事务在同一条件名称上执行 NOTIFY 然后执行 LISTEN，它将在提交时收到自我通知。这有点奇怪，但与我们的历史行为是一致的。
 */
typedef struct Notification
{
	uint16		channel_len;	/* 通道名称字符串的长度 */
	uint16		payload_len;	/* 有效载荷字符串的长度 */
	/* null 终止的通道名称，然后是 null 终止的有效载荷 */
	char		data[FLEXIBLE_ARRAY_MEMBER];
} Notification;

typedef struct NotificationList
{
	int			nestingLevel;	/* 当前事务嵌套深度 */
	List	   *events;			/* Notification 结构的列表 */
	HTAB	   *hashtab;		/* NotificationHash 结构的哈希，或 NULL */
	struct NotificationList *upper; /* 上层事务级别的详细信息 */
} NotificationList;

#define MIN_HASHABLE_NOTIFIES 16	/* 构建哈希表的阈值 */

typedef struct NotificationHash
{
	Notification *event;		/* => 实际的 Notification 结构 */
} NotificationHash;

static NotificationList *pendingNotifies = NULL;

/*
 * 入站通知最初由 HandleNotifyInterrupt() 处理，该函数在信号处理程序内部被调用。它只是设置 notifyInterruptPending 标志并设置进程的等待标志。随后，ProcessNotifyInterrupt() 将在安全处理中断时被调用。
 */
volatile sig_atomic_t notifyInterruptPending = false;

/* 如果我们已经注册了 on_shmem_exit 清理则为真 */
static bool unlistenExitRegistered = false;

/* 如果我们当前注册为 asyncQueueControl 中的监听者，则为真 */
static bool amRegisteredListener = false;

/* 我们是否已经进展到一个是 QUEUE_CLEANUP_DELAY 的倍数的页面？ */
static bool tryAdvanceTail = false;

/* GUC 参数 */
bool		Trace_notify = false;

/* 本地函数原型 */
static int	fc_asyncQueuePageDiff(int fc_p, int fc_q);
static bool fc_asyncQueuePagePrecedes(int fc_p, int fc_q);
static void fc_queue_listen(ListenActionKind fc_action, const char *fc_channel);
static void fc_Async_UnlistenOnExit(int fc_code, Datum fc_arg);
static void fc_Exec_ListenPreCommit(void);
static void fc_Exec_ListenCommit(const char *fc_channel);
static void fc_Exec_UnlistenCommit(const char *fc_channel);
static void fc_Exec_UnlistenAllCommit(void);
static bool fc_IsListeningOn(const char *fc_channel);
static void fc_asyncQueueUnregister(void);
static bool fc_asyncQueueIsFull(void);
static bool fc_asyncQueueAdvance(volatile QueuePosition *fc_position, int fc_entryLength);
static void fc_asyncQueueNotificationToEntry(Notification *fc_n, AsyncQueueEntry *fc_qe);
static ListCell *fc_asyncQueueAddEntries(ListCell *fc_nextNotify);
static double fc_asyncQueueUsage(void);
static void fc_asyncQueueFillWarning(void);
static void fc_SignalBackends(void);
static void fc_asyncQueueReadAllNotifications(void);
static bool fc_asyncQueueProcessPageEntries(volatile QueuePosition *fc_current,
										 QueuePosition fc_stop,
										 char *fc_page_buffer,
										 Snapshot fc_snapshot);
static void fc_asyncQueueAdvanceTail(void);
static void fc_ProcessIncomingNotify(bool fc_flush);
static bool fc_AsyncExistsPendingNotify(Notification *fc_n);
static void fc_AddEventToPendingNotifies(Notification *fc_n);
static uint32 fc_notification_hash(const void *fc_key, Size fc_keysize);
static int	fc_notification_match(const void *fc_key1, const void *fc_key2, Size fc_keysize);
static void fc_ClearPendingActionsAndNotifies(void);

/*
 * 计算两个队列页面编号之间的差（即，p - q），
 * 考虑到回绕情况。
 */
static int fc_asyncQueuePageDiff(int fc_p, int fc_q)
{
	int			fc_diff;

	/*
	 * 我们必须比较模（QUEUE_MAX_PAGE+1）/2。两个输入应该在
	 * 范围 0..QUEUE_MAX_PAGE 之间。
	 */
	Assert(fc_p >= 0 && fc_p <= QUEUE_MAX_PAGE);
	Assert(fc_q >= 0 && fc_q <= QUEUE_MAX_PAGE);

	fc_diff = fc_p - fc_q;
	if (fc_diff >= ((QUEUE_MAX_PAGE + 1) / 2))
		fc_diff -= QUEUE_MAX_PAGE + 1;
	else if (fc_diff < -((QUEUE_MAX_PAGE + 1) / 2))
		fc_diff += QUEUE_MAX_PAGE + 1;
	return fc_diff;
}

/*
 * p < q 包括回绕吗？
 *
 * 由于 asyncQueueIsFull() 阻止创建可能在任何
 * 现有页面之前的页面，我们不需要评估一个页面内的条目。
 */
static bool fc_asyncQueuePagePrecedes(int fc_p, int fc_q)
{
	return fc_asyncQueuePageDiff(fc_p, fc_q) < 0;
}

/*
 * 报告我们共享内存区域所需的空间
 */
Size AsyncShmemSize(void)
{
	Size		fc_size;

	/* 这最好与 AsyncShmemInit 匹配 */
	fc_size = mul_size(MaxBackends + 1, sizeof(QueueBackendStatus));
	fc_size = add_size(fc_size, offsetof(AsyncQueueControl, backend));

	fc_size = add_size(fc_size, SimpleLruShmemSize(NUM_NOTIFY_BUFFERS, 0));

	return fc_size;
}

/*
 * 初始化我们的共享内存区域
 */
void AsyncShmemInit(void)
{
	bool		fc_found;
	Size		fc_size;

	/*
	 * 创建或附加到 AsyncQueueControl 结构。
	 *
	 * 后台中使用的条目[]数组从 1 到 MaxBackends；第零个条目未使用但
	 * 必须分配。
	 */
	fc_size = mul_size(MaxBackends + 1, sizeof(QueueBackendStatus));
	fc_size = add_size(fc_size, offsetof(AsyncQueueControl, backend));

	asyncQueueControl = (AsyncQueueControl *)
		ShmemInitStruct("Async Queue Control", fc_size, &fc_found);

	if (!fc_found)
	{
		/* 第一次通过，所以初始化它 */
		SET_QUEUE_POS(QUEUE_HEAD, 0, 0);
		SET_QUEUE_POS(QUEUE_TAIL, 0, 0);
		QUEUE_STOP_PAGE = 0;
		QUEUE_FIRST_LISTENER = InvalidBackendId;
		asyncQueueControl->lastQueueFillWarn = 0;
		/* 零个条目不会被使用，但还是让我们初始化它吧 */
		for (int fc_i = 0; fc_i <= MaxBackends; fc_i++)
		{
			QUEUE_BACKEND_PID(fc_i) = InvalidPid;
			QUEUE_BACKEND_DBOID(fc_i) = InvalidOid;
			QUEUE_NEXT_LISTENER(fc_i) = InvalidBackendId;
			SET_QUEUE_POS(QUEUE_BACKEND_POS(fc_i), 0, 0);
		}
	}

	/*
	 * 设置 pg_notify 数据的 SLRU 管理。
	 */
	NotifyCtl->PagePrecedes = fc_asyncQueuePagePrecedes;
	SimpleLruInit(NotifyCtl, "Notify", NUM_NOTIFY_BUFFERS, 0,
				  NotifySLRULock, "pg_notify", LWTRANCHE_NOTIFY_BUFFER,
				  SYNC_HANDLER_NONE);

	if (!fc_found)
	{
		/*
		 * 在启动或重启期间，清空 pg_notify 目录。
		 */
		(void) SlruScanDirectory(NotifyCtl, SlruScanDirCbDeleteAll, NULL);
	}
}


/*
 * pg_notify -
 *	  发送通知事件的 SQL 函数
 */
Datum pg_notify(PG_FUNCTION_ARGS)
{
	const char *fc_channel;
	const char *fc_payload;

	if (PG_ARGISNULL(0))
		fc_channel = "";
	else
		fc_channel = text_to_cstring(PG_GETARG_TEXT_PP(0));

	if (PG_ARGISNULL(1))
		fc_payload = "";
	else
		fc_payload = text_to_cstring(PG_GETARG_TEXT_PP(1));

	/* 对于作为语句的 NOTIFY，这在 ProcessUtility 中检查 */
	PreventCommandDuringRecovery("NOTIFY");

	Async_Notify(fc_channel, fc_payload);

	PG_RETURN_VOID();
}


/*
 * Async_Notify
 *
 *		这由 SQL 通知命令执行。
 *
 *		将消息添加到待处理通知的列表中。
 *		实际的通知发生在事务提交时。
 *		^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 */
void Async_Notify(const char *fc_channel, const char *fc_payload)
{
	int			fc_my_level = GetCurrentTransactionNestLevel();
	size_t		fc_channel_len;
	size_t		fc_payload_len;
	Notification *fc_n;
	MemoryContext fc_oldcontext;

	if (IsParallelWorker())
		elog(ERROR, "cannot send notifications from a parallel worker");

	if (Trace_notify)
		elog(DEBUG1, "Async_Notify(%s)", fc_channel);

	fc_channel_len = fc_channel ? strlen(fc_channel) : 0;
	fc_payload_len = fc_payload ? strlen(fc_payload) : 0;

	/* 必须指定一个通道名称 */
	if (fc_channel_len == 0)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("channel name cannot be empty")));

	/* 强制长度限制 */
	if (fc_channel_len >= NAMEDATALEN)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("channel name too long")));

	if (fc_payload_len >= NOTIFY_PAYLOAD_MAX_LENGTH)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("payload string too long")));

	/*
	 * 我们必须构造通知条目，即使最终不使用
	 * 它，也要便于与现有列表条目进行比较。
	 *
	 * 通知列表需要存活到事务结束，因此将其存储
	 * 在事务上下文中。
	 */
	fc_oldcontext = MemoryContextSwitchTo(CurTransactionContext);

	fc_n = (Notification *) palloc(offsetof(Notification, data) +
								fc_channel_len + fc_payload_len + 2);
	fc_n->channel_len = fc_channel_len;
	fc_n->payload_len = fc_payload_len;
	strcpy(fc_n->data, fc_channel);
	if (fc_payload)
		strcpy(fc_n->data + fc_channel_len + 1, fc_payload);
	else
		fc_n->data[fc_channel_len + 1] = '\0';

	if (pendingNotifies == NULL || fc_my_level > pendingNotifies->nestingLevel)
	{
		NotificationList *fc_notifies;

		/*
		 * 当前 (sub)xact 中的第一个通知事件。注意我们在 TopTransactionContext 中分配了
		 * NotificationList；嵌套级别可能会在 AtSubCommit_Notify 中稍后改变。
		 */
		fc_notifies = (NotificationList *)
			MemoryContextAlloc(TopTransactionContext,
							   sizeof(NotificationList));
		fc_notifies->nestingLevel = fc_my_level;
		fc_notifies->events = list_make1(fc_n);
		/* 我们当然现在不需要哈希表 */
		fc_notifies->hashtab = NULL;
		fc_notifies->upper = pendingNotifies;
		pendingNotifies = fc_notifies;
	}
	else
	{
		/* 现在检查重复项 */
		if (fc_AsyncExistsPendingNotify(fc_n))
		{
			/* 这是一个重复项，所以忘记它 */
			pfree(fc_n);
			MemoryContextSwitchTo(fc_oldcontext);
			return;
		}

		/* 将更多事件附加到现有列表中 */
		fc_AddEventToPendingNotifies(fc_n);
	}

	MemoryContextSwitchTo(fc_oldcontext);
}

/*
 * queue_listen
 *		listen、unlisten 和 unlisten all 命令的公共代码。
 *
 *		将请求添加到待处理操作的列表中。
 *		listenChannels 列表的实际更新发生在事务
 *		提交期间。
 */
static void fc_queue_listen(ListenActionKind fc_action, const char *fc_channel)
{
	MemoryContext fc_oldcontext;
	ListenAction *fc_actrec;
	int			fc_my_level = GetCurrentTransactionNestLevel();

	/*
	 * 与 Async_Notify 不同，我们不尝试去合并重复项。确保
	 * 我们获得冲突的 LISTEN/UNLISTEN/UNLISTEN_ALL 的正确交互将
	 * 太复杂，而且在合理的应用程序中不太可能有任何性能
	 * 提升。
	 */
	fc_oldcontext = MemoryContextSwitchTo(CurTransactionContext);

	/* 为终止空字符保留空间，包括在sizeof(ListenAction)中 */
	fc_actrec = (ListenAction *) palloc(offsetof(ListenAction, channel) +
									 strlen(fc_channel) + 1);
	fc_actrec->action = fc_action;
	strcpy(fc_actrec->channel, fc_channel);

	if (pendingActions == NULL || fc_my_level > pendingActions->nestingLevel)
	{
		ActionList *fc_actions;

		/*
		 * 当前子事务中的第一个操作。注意我们在TopTransactionContext中分配
		 * ActionList；嵌套级别可能会通过AtSubCommit_Notify稍后更改。
		 */
		fc_actions = (ActionList *)
			MemoryContextAlloc(TopTransactionContext, sizeof(ActionList));
		fc_actions->nestingLevel = fc_my_level;
		fc_actions->actions = list_make1(fc_actrec);
		fc_actions->upper = pendingActions;
		pendingActions = fc_actions;
	}
	else
		pendingActions->actions = lappend(pendingActions->actions, fc_actrec);

	MemoryContextSwitchTo(fc_oldcontext);
}

/*
 * Async_Listen
 *
 *		这是由SQL监听命令执行的。
 */
void Async_Listen(const char *fc_channel)
{
	if (Trace_notify)
		elog(DEBUG1, "Async_Listen(%s,%d)", fc_channel, MyProcPid);

	fc_queue_listen(LISTEN_LISTEN, fc_channel);
}

/*
 * Async_Unlisten
 *
 *		这是由SQL取消监听命令执行的。
 */
void Async_Unlisten(const char *fc_channel)
{
	if (Trace_notify)
		elog(DEBUG1, "Async_Unlisten(%s,%d)", fc_channel, MyProcPid);

	/* 如果我们根本不可能在监听，就不需要排队任何内容 */
	if (pendingActions == NULL && !unlistenExitRegistered)
		return;

	fc_queue_listen(LISTEN_UNLISTEN, fc_channel);
}

/*
 * Async_UnlistenAll
 *
 *		这是由UNLISTEN *命令调用的，也在后端退出时调用。
 */
void Async_UnlistenAll(void)
{
	if (Trace_notify)
		elog(DEBUG1, "Async_UnlistenAll(%d)", MyProcPid);

	/* 如果我们根本不可能在监听，就不需要排队任何内容 */
	if (pendingActions == NULL && !unlistenExitRegistered)
		return;

	fc_queue_listen(LISTEN_UNLISTEN_ALL, "");
}

/*
 * SQL函数：返回该后端正在积极监听的频道名称的集合。
 *
 * 注意：此编码依赖于监听频道列表在事务内无法更改的事实。
 */
Datum pg_listening_channels(PG_FUNCTION_ARGS)
{
	FuncCallContext *fc_funcctx;

	/* 仅在函数的第一次调用时完成的工作 */
	if (SRF_IS_FIRSTCALL())
	{
		/* 为跨调用持久性创建一个函数上下文 */
		fc_funcctx = SRF_FIRSTCALL_INIT();
	}

	/* 每次调用函数时完成的工作 */
	fc_funcctx = SRF_PERCALL_SETUP();

	if (fc_funcctx->call_cntr < list_length(listenChannels))
	{
		char	   *fc_channel = (char *) list_nth(listenChannels,
												fc_funcctx->call_cntr);

		SRF_RETURN_NEXT(fc_funcctx, CStringGetTextDatum(fc_channel));
	}

	SRF_RETURN_DONE(fc_funcctx);
}

/*
 * Async_UnlistenOnExit
 *
 * 如果我们在此后端执行了任何LISTEN操作，则在后端退出时执行此操作。
 * 如果用户取消监听了所有内容，这可能不再必要，但我们不尝试检测该情况。
 */
static void fc_Async_UnlistenOnExit(int fc_code, Datum fc_arg)
{
	fc_Exec_UnlistenAllCommit();
	fc_asyncQueueUnregister();
}

/*
 * AtPrepare_Notify
 *
 *		在两阶段事务的准备阶段调用此函数。保存状态以备稍后提交。
 */
void AtPrepare_Notify(void)
{
	/* 不允许有任何挂起的LISTEN/UNLISTEN/NOTIFY操作 */
	if (pendingActions || pendingNotifies)
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("cannot PREPARE a transaction that has executed LISTEN, UNLISTEN, or NOTIFY")));
}

/*
 * PreCommit_Notify
 *
 *		在事务提交时调用此函数，在实际提交到clog之前。
 *
 *		如果有挂起的LISTEN操作，请确保我们在共享内存监听器数组中
 *		被列出。这必须在提交之前发生，以确保我们不会错过任何
 *		在我们之后提交的事务的通知。
 *
 *		如果在pendingNotifies列表中有出站通知请求，请将它们添加到
 *		全局队列。我们在提交之前这样做，以便在队列空间不足时仍然
 *		可以抛出错误。
 */
void PreCommit_Notify(void)
{
	ListCell   *fc_p;

	if (!pendingActions && !pendingNotifies)
		return;					/* 此事务中没有相关语句 */

	if (Trace_notify)
		elog(DEBUG1, "PreCommit_Notify");

	/* 对任何挂起的监听/取消监听操作进行预检查 */
	if (pendingActions != NULL)
	{
		foreach(fc_p, pendingActions->actions)
		{
			ListenAction *fc_actrec = (ListenAction *) lfirst(fc_p);

			switch (fc_actrec->action)
			{
				case LISTEN_LISTEN:
					fc_Exec_ListenPreCommit();
					break;
				case LISTEN_UNLISTEN:
					/* 没有Exec_UnlistenPreCommit() */
					break;
				case LISTEN_UNLISTEN_ALL:
					/* 没有Exec_UnlistenAllPreCommit() */
					break;
			}
		}
	}

	/* 队列任何挂起的通知（必须在上述操作后发生） */
	if (pendingNotifies)
	{
		ListCell   *fc_nextNotify;

		/*
		 * 确保我们为当前事务分配了XID。
		 * 如果我们已经有一个XID，GetCurrentTransactionId是便宜的，
		 * 但如果没有就不那么便宜了，并且我们更希望在持有
		 * NotifyQueueLock时不执行那项工作。
		 */
		(void) GetCurrentTransactionId();

		/*
		 * 通过获取一个特殊的锁来序列化写入者，直到提交之后我们才释放这个锁。
		 * 这确保了队列条目按提交顺序出现，特别是确保不存在未提交的队列条目在已提交条目之前，因此未提交的事务无法阻塞可交付通知的发送。
		 *
		 * 我们使用重量级锁，这样在提交或中止后会自动释放。这也允许检测死锁，尽管在这里真的不应该发生死锁。
		 *
		 * 锁是在“数据库 0”上，这看起来相当糟糕，但似乎不值得为了这个发明一个特殊的锁标记类别。
		 * （历史注释：在 PG 9.0 之前，“数据库 0”上的类似锁是由 flatfiles 机制使用的。）
		 */
		LockSharedObject(DatabaseRelationId, InvalidOid, 0,
						 AccessExclusiveLock);

		/* 现在将通知推送到队列中 */
		fc_nextNotify = list_head(pendingNotifies->events);
		while (fc_nextNotify != NULL)
		{
			/*
			 * 将待定通知添加到队列中。我们每页获取并释放 NotifyQueueLock 一次，这可能有些过头，
			 * 但确实允许读取者在我们进行此操作时进入。
			 *
			 * 完整的队列是非常不常见的，真的不应该发生，因为在 SLRU 页中有这么多可用空间。
			 * 尽管如此，我们仍然需要处理这种可能性。请注意，当我们到达这里时，我们正在提交我们的
			 * 事务，但我们尚未提交到 clog，因此在这一时刻我们仍然可以回滚事务。
			 */
			LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
			fc_asyncQueueFillWarning();
			if (fc_asyncQueueIsFull())
				ereport(ERROR,
						(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
						 errmsg("too many notifications in the NOTIFY queue")));
			fc_nextNotify = fc_asyncQueueAddEntries(fc_nextNotify);
			LWLockRelease(NotifyQueueLock);
		}

		/* 请注意，我们不清除 pendingNotifies；AtCommit_Notify 会清除。 */
	}
}

/*
 * AtCommit_Notify
 *
 *		这是在事务提交时调用的，在提交到 clog 之后。
 *
 *		更新 listenChannels 并清除事务局部状态。
 *
 *		如果我们在事务中发出了任何通知，则向正在收听的后端（可能包括我们自己）发送信号以处理它们。
 *		此外，如果我们用新的通知填满了足够的队列页面，则尝试推进队列尾指针。
 */
void AtCommit_Notify(void)
{
	ListCell   *fc_p;

	/*
	 * 允许未执行 LISTEN/UNLISTEN/NOTIFY 的事务尽快返回
	 */
	if (!pendingActions && !pendingNotifies)
		return;

	if (Trace_notify)
		elog(DEBUG1, "AtCommit_Notify");

	/* 执行任何待定的 listen/unlisten 操作 */
	if (pendingActions != NULL)
	{
		foreach(fc_p, pendingActions->actions)
		{
			ListenAction *fc_actrec = (ListenAction *) lfirst(fc_p);

			switch (fc_actrec->action)
			{
				case LISTEN_LISTEN:
					fc_Exec_ListenCommit(fc_actrec->channel);
					break;
				case LISTEN_UNLISTEN:
					fc_Exec_UnlistenCommit(fc_actrec->channel);
					break;
				case LISTEN_UNLISTEN_ALL:
					fc_Exec_UnlistenAllCommit();
					break;
			}
		}
	}

	/* 如果不再收听任何内容，则从监听器数组中退出 */
	if (amRegisteredListener && listenChannels == NIL)
		fc_asyncQueueUnregister();

	/*
	 * 向正在收听的后端发送信号。只有在有待处理的通知时我们才需要这样做，
	 * 这些通知之前已由 PreCommit_Notify() 添加到共享队列中。
	 */
	if (pendingNotifies != NULL)
		fc_SignalBackends();

	/*
	 * 如果是时候尝试推进全局尾指针，则执行此操作。
	 *
	 * （在发送方执行此操作可能看起来奇怪，因为更有可能的是监听者尚未读取我们刚刚发送的消息。
	 * 但是，如果只有发送者执行此操作，则争用会减少，而且推进全局尾部并没有太大的紧迫性。
	 * 因此，这通常将清理一些早先发送的消息。）
	 */
	if (tryAdvanceTail)
	{
		tryAdvanceTail = false;
		fc_asyncQueueAdvanceTail();
	}

	/* 并进行清理 */
	fc_ClearPendingActionsAndNotifies();
}

/*
 * Exec_ListenPreCommit --- PreCommit_Notify的子例程
 *
 * 此函数必须确保我们准备好接收任何传入的消息。
 */
static void fc_Exec_ListenPreCommit(void)
{
	QueuePosition fc_head;
	QueuePosition fc_max;
	BackendId	fc_prevListener;

	/*
	 * 如果我们已经在监听某些内容，或者在该事务中已经运行过此例程，则无需执行任何操作。
	 */
	if (amRegisteredListener)
		return;

	if (Trace_notify)
		elog(DEBUG1, "Exec_ListenPreCommit(%d)", MyProcPid);

	/*
	 * 在注册之前，确保我们在终止之前会取消监听。（注意：
	 * 如果我们稍后中止，此操作不会被撤销。）
	 */
	if (!unlistenExitRegistered)
	{
		before_shmem_exit(fc_Async_UnlistenOnExit, 0);
		unlistenExitRegistered = true;
	}

	/*
	 * 这是我们的第一次LISTEN，因此建立我们的指针。
	 *
	 * 我们将指针设置为全局尾指针，然后将其向前移动
	 * 经过已提交的通知。这确保我们不会错过任何
	 * 尚未提交的通知。我们可能会再收到一些，但这
	 * 并没有坏处。
	 *
	 * 在某些情况下，可能会有很多已提交的通知
	 * 尚未被修剪（因为某个后端在读取它们时很懒）。
	 * 为了减少我们的启动时间，我们可以查看其他
	 * 后端，并采用在我们数据库中任何后端的最大“pos”指针；
	 * 它已经提前过的任何通知肯定是已提交的，因此我们无需
	 * 再次检查。 （我们必须只考虑连接到我们数据库的
	 * 后端，因为其他后端不会关心检查我们数据库中通知的
	 * 提交状态。）
	 *
	 * 我们这里需要独占锁，以便查看其他后端的条目
	 * 并操作列表链接。
	 */
	LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
	fc_head = QUEUE_HEAD;
	fc_max = QUEUE_TAIL;
	fc_prevListener = InvalidBackendId;
	for (BackendId fc_i = QUEUE_FIRST_LISTENER; fc_i > 0; fc_i = QUEUE_NEXT_LISTENER(fc_i))
	{
		if (QUEUE_BACKEND_DBOID(fc_i) == MyDatabaseId)
			fc_max = QUEUE_POS_MAX(fc_max, QUEUE_BACKEND_POS(fc_i));
		/* 还要找到此后端之前最后一个正在监听的后端 */
		if (fc_i < MyBackendId)
			fc_prevListener = fc_i;
	}
	QUEUE_BACKEND_POS(MyBackendId) = fc_max;
	QUEUE_BACKEND_PID(MyBackendId) = MyProcPid;
	QUEUE_BACKEND_DBOID(MyBackendId) = MyDatabaseId;
	/* 在正确的位置将后端插入到监听者列表中 */
	if (fc_prevListener > 0)
	{
		QUEUE_NEXT_LISTENER(MyBackendId) = QUEUE_NEXT_LISTENER(fc_prevListener);
		QUEUE_NEXT_LISTENER(fc_prevListener) = MyBackendId;
	}
	else
	{
		QUEUE_NEXT_LISTENER(MyBackendId) = QUEUE_FIRST_LISTENER;
		QUEUE_FIRST_LISTENER = MyBackendId;
	}
	LWLockRelease(NotifyQueueLock);

	/* 现在我们在全局数组中被列出，因此记住我们正在监听 */
	amRegisteredListener = true;

	/*
	 * 尝试尽可能向前移动我们的指针。这将跳过
	 * 已提交的通知，我们希望这样做，因为它们
	 * 可能相当过时。注意，我们还没有在监听任何内容，
	 * 因此我们不会将这些通知发送到我们的前端。此外，
	 * 虽然我们的事务可能执行了NOTIFY，但这些消息尚未
	 * 入队，因此我们不会在这里跳过它们。
	 */
	if (!QUEUE_POS_EQUAL(fc_max, fc_head))
		fc_asyncQueueReadAllNotifications();
}

/*
 * Exec_ListenCommit --- AtCommit_Notify的子例程
 *
 * 将频道添加到我们正在监听的频道列表中。
 */
static void fc_Exec_ListenCommit(const char *fc_channel)
{
	MemoryContext fc_oldcontext;

	/* 如果我们已经在监听此频道，则不执行任何操作 */
	if (fc_IsListeningOn(fc_channel))
		return;

	/*
	 * 将新频道名称添加到listenChannels中。
	 *
	 * XXX 理论上这里可能会出现内存不足的错误，
	 * 这将很糟糕，因为我们已经提交。此时似乎不值得尝试
	 * 进行防范，但可能以后改善这一点。
	 */
	fc_oldcontext = MemoryContextSwitchTo(TopMemoryContext);
	listenChannels = lappend(listenChannels, pstrdup(fc_channel));
	MemoryContextSwitchTo(fc_oldcontext);
}

/*
 * Exec_UnlistenCommit --- AtCommit_Notify的子例程
 *
 * 从listenChannels中移除指定的频道名称。
 */
static void fc_Exec_UnlistenCommit(const char *fc_channel)
{
	ListCell   *fc_q;

	if (Trace_notify)
		elog(DEBUG1, "Exec_UnlistenCommit(%s,%d)", fc_channel, MyProcPid);

	foreach(fc_q, listenChannels)
	{
		char	   *fc_lchan = (char *) lfirst(fc_q);

		if (strcmp(fc_lchan, fc_channel) == 0)
		{
			listenChannels = foreach_delete_current(listenChannels, fc_q);
			pfree(fc_lchan);
			break;
		}
	}

	/*
	 * 我们不会对取消监听未被监听的内容表示抱怨；
	 * 我们应该这样做吗？
	 */
}

/*
 * Exec_UnlistenAllCommit --- AtCommit_Notify的子例程
 *
 *		取消此后端在所有频道上的监听。
 */
static void fc_Exec_UnlistenAllCommit(void)
{
	if (Trace_notify)
		elog(DEBUG1, "Exec_UnlistenAllCommit(%d)", MyProcPid);

	list_free_deep(listenChannels);
	listenChannels = NIL;
}

/*
 * 测试我们是否在给定的通道名称上主动监听。
 *
 * 注意：此函数会对队列中找到的每个通知执行。
 * 可能值得进一步优化，例如将列表转换为排序数组，以便可以进行二分查找。
 * 实际上，列表可能相对较短。
 */
static bool fc_IsListeningOn(const char *fc_channel)
{
	ListCell   *fc_p;

	foreach(fc_p, listenChannels)
	{
		char	   *fc_lchan = (char *) lfirst(fc_p);

		if (strcmp(fc_lchan, fc_channel) == 0)
			return true;
	}
	return false;
}

/*
 * 当我们不再监听任何通道时，从监听器数组中移除我们的条目。
 * 注意：如果我们已经不在监听，必须不失败。
 */
static void fc_asyncQueueUnregister(void)
{
	Assert(listenChannels == NIL);	/*
		 * 我们使用 CaseTestExpr 作为强制转换树的基础，
		 * 因为插入源值是非常便宜的。
		 */

	if (!amRegisteredListener)	/* 无需操作 */
		return;

	/*
	 * 在这里需要独占锁以操作列表链接。
	 */
	LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
	/* 将我们的条目标记为无效 */
	QUEUE_BACKEND_PID(MyBackendId) = InvalidPid;
	QUEUE_BACKEND_DBOID(MyBackendId) = InvalidOid;
	/* 并将其从列表中移除 */
	if (QUEUE_FIRST_LISTENER == MyBackendId)
		QUEUE_FIRST_LISTENER = QUEUE_NEXT_LISTENER(MyBackendId);
	else
	{
		for (BackendId fc_i = QUEUE_FIRST_LISTENER; fc_i > 0; fc_i = QUEUE_NEXT_LISTENER(fc_i))
		{
			if (QUEUE_NEXT_LISTENER(fc_i) == MyBackendId)
			{
				QUEUE_NEXT_LISTENER(fc_i) = QUEUE_NEXT_LISTENER(MyBackendId);
				break;
			}
		}
	}
	QUEUE_NEXT_LISTENER(MyBackendId) = InvalidBackendId;
	LWLockRelease(NotifyQueueLock);

	/* 将自己标记为不再列在全局数组中 */
	amRegisteredListener = false;
}

/*
 * 测试是否有空间插入更多通知消息。
 *
 * 调用者必须至少持有共享的 NotifyQueueLock。
 */
static bool fc_asyncQueueIsFull(void)
{
	int			fc_nexthead;
	int			fc_boundary;

	/*
	 * 如果创建一个新的头页面将创建一个在逻辑上位于当前全局尾指针之前的页面，
	 * 则队列已满，即头指针将与尾指针环绕。
	 * 我们不能创建这样的头页面以免混淆 slru.c。
	 * 出于安全考虑，我们将尾指针向后舍入到段边界（
	 * asyncQueueAdvanceTail 中的截断逻辑不执行此操作，因此在这里执行是可选的）。
	 *
	 * 请注意，这一测试*不*依赖于当前头页面上有多少空间。
	 * 这是必要的，因为 asyncQueueAddEntries 可能会尝试创建下一个头页面。
	 */
	fc_nexthead = QUEUE_POS_PAGE(QUEUE_HEAD) + 1;
	if (fc_nexthead > QUEUE_MAX_PAGE)
		fc_nexthead = 0;			/* 环绕 */
	fc_boundary = QUEUE_STOP_PAGE;
	fc_boundary -= fc_boundary % SLRU_PAGES_PER_SEGMENT;
	return fc_asyncQueuePagePrecedes(fc_nexthead, fc_boundary);
}

/*
 * 将队列位置向前推进到下一个条目，假设当前条目的长度为 entryLength。
 * 如果我们跳到一个新页面，则函数返回 true，否则返回 false。
 */
static bool fc_asyncQueueAdvance(volatile QueuePosition *fc_position, int fc_entryLength)
{
	int			fc_pageno = QUEUE_POS_PAGE(*fc_position);
	int			fc_offset = QUEUE_POS_OFFSET(*fc_position);
	bool		fc_pageJump = false;

	/*
	 * 移动到下一个写入位置：首先跳过我们刚刚写入或读取的内容。
	 */
	fc_offset += fc_entryLength;
	Assert(fc_offset <= QUEUE_PAGESIZE);

	/*
	 * 第二步，检查页面上是否可以写入另一个条目。
	 * 如果可以，就停留在这里，我们已到达下一个位置。
	 * 如果不可以，则需要转到下一个页面。
	 */
	if (fc_offset + QUEUEALIGN(AsyncQueueEntryEmptySize) > QUEUE_PAGESIZE)
	{
		fc_pageno++;
		if (fc_pageno > QUEUE_MAX_PAGE)
			fc_pageno = 0;			/* 环绕 */
		fc_offset = 0;
		fc_pageJump = true;
	}

	SET_QUEUE_POS(*fc_position, fc_pageno, fc_offset);
	return fc_pageJump;
}

/*
 * 在 *qe 中填充 AsyncQueueEntry，填入出站通知消息。
 */
static void fc_asyncQueueNotificationToEntry(Notification *fc_n, AsyncQueueEntry *fc_qe)
{
	size_t		fc_channellen = fc_n->channel_len;
	size_t		fc_payloadlen = fc_n->payload_len;
	int			fc_entryLength;

	Assert(fc_channellen < NAMEDATALEN);
	Assert(fc_payloadlen < NOTIFY_PAYLOAD_MAX_LENGTH);

	/* 终止符已包含在 AsyncQueueEntryEmptySize 中 */
	fc_entryLength = AsyncQueueEntryEmptySize + fc_payloadlen + fc_channellen;
	fc_entryLength = QUEUEALIGN(fc_entryLength);
	fc_qe->length = fc_entryLength;
	fc_qe->dboid = MyDatabaseId;
	fc_qe->xid = GetCurrentTransactionId();
	fc_qe->srcPid = MyProcPid;
	memcpy(fc_qe->data, fc_n->data, fc_channellen + fc_payloadlen + 2);
}

/*
 * 将待定通知添加到队列中。
 *
 * 我们在这里按页面逐步处理，即一旦需要转到新页面就会停止，
 * 但我们会再次被调用，然后填充下一页。
 * 如果条目无法适应当前页面，我们会写入一个虚拟条目，
 * 其数据库 OID 为 InvalidOid，以便填满页面。
 * 因此每个页面总是使用到最后一个字节，这简化了后续读取页面的过程。
 *
 * 我们正在处理包含下一个要写入的通知的列表单元（在 pendingNotifies->events 中），
 * 并返回第一个尚未写入的单元。
 * 最终我们将返回 NULL，表示所有操作完成。
 *
 * 我们已经从调用者那里持有了 NotifyQueueLock，并在此函数中本地抓取了 NotifySLRULock。
 */
static ListCell *
fc_asyncQueueAddEntries(ListCell *fc_nextNotify)
{
	AsyncQueueEntry fc_qe;
	QueuePosition fc_queue_head;
	int			fc_pageno;
	int			fc_offset;
	int			fc_slotno;

	/* 我们在此操作期间同时持有 NotifyQueueLock 和 NotifySLRULock */
	LWLockAcquire(NotifySLRULock, LW_EXCLUSIVE);

	/*
	 * 我们使用 QUEUE_HEAD 的本地副本，在退出时将其写回共享
	 * 内存。这样做的原因是，如果我们必须推进到新页面，SimpleLruZeroPage 可能会失败（例如，磁盘空间不足），如果失败，我们绝不能推进 QUEUE_HEAD。（否则，后续插入将尝试将条目放入 slru.c 认为尚不存在的页面。）因此，使用一个本地位置变量。请注意，如果我们失败，任何已经插入的队列条目都会被遗忘；这没关系，因为在我们的事务回滚后，它们无论如何都是无用的。
	 */
	fc_queue_head = QUEUE_HEAD;

	/*
	 * 如果这是自从 postmaster 启动以来的第一次写入，我们需要
	 * 初始化异步 SLRU 的第一页。否则，当前页面应该已被初始化，因此只需获取它。
	 *
	 * （当 SLRU 位置刚刚回绕时，我们也可以采取第一条路径，但在这种情况下重新归零页面是无害的。）
	 */
	fc_pageno = QUEUE_POS_PAGE(fc_queue_head);
	if (QUEUE_POS_IS_ZERO(fc_queue_head))
		fc_slotno = SimpleLruZeroPage(NotifyCtl, fc_pageno);
	else
		fc_slotno = SimpleLruReadPage(NotifyCtl, fc_pageno, true,
								   InvalidTransactionId);

	/* 注意，在写入之前我们将页面标记为脏 */
	NotifyCtl->shared->page_dirty[fc_slotno] = true;

	while (fc_nextNotify != NULL)
	{
		Notification *fc_n = (Notification *) lfirst(fc_nextNotify);

		/* 在本地变量 qe 中构造一个有效的队列条目 */
		fc_asyncQueueNotificationToEntry(fc_n, &fc_qe);

		fc_offset = QUEUE_POS_OFFSET(fc_queue_head);

		/* 检查条目是否真的适合当前页面 */
		if (fc_offset + fc_qe.length <= QUEUE_PAGESIZE)
		{
			/* 好的，所以将 nextNotify 向前推进到这个项目之后 */
			fc_nextNotify = lnext(pendingNotifies->events, fc_nextNotify);
		}
		else
		{
			/*
			 * 写入一个虚拟条目以填满页面。实际上，读取者只会检查 dboid，
			 * 由于它不会匹配任何读取者的数据库 OID，他们会忽略此条目并继续。
			 */
			fc_qe.length = QUEUE_PAGESIZE - fc_offset;
			fc_qe.dboid = InvalidOid;
			fc_qe.data[0] = '\0';	/* 空通道 */
			fc_qe.data[1] = '\0';	/* 空有效载荷 */
		}

		/* 现在将 qe 复制到共享缓冲区页面 */
		memcpy(NotifyCtl->shared->page_buffer[fc_slotno] + fc_offset,
			   &fc_qe,
			   fc_qe.length);

		/* 适当地推进 queue_head，并检测页面是否已满 */
		if (fc_asyncQueueAdvance(&(fc_queue_head), fc_qe.length))
		{
			/*
			 * 页面已满，所以我们在这里完成，但首先用零填充下一页。
			 * 这样做的原因是确保 slru.c 对头页面的理解始终与我们的相同，
			 * 这避免了在 SimpleLruTruncate 中出现边界问题。函数
			 * asyncQueueIsFull() 确保可以创建此页面而不会溢出队列。
			 */
			fc_slotno = SimpleLruZeroPage(NotifyCtl, QUEUE_POS_PAGE(fc_queue_head));

			/*
			 * 如果新页面地址是 QUEUE_CLEANUP_DELAY 的倍数，
			 * 设置标志以记住我们应该尝试推进尾指针（我们不想在这里实际执行此操作）。
			 */
			if (QUEUE_POS_PAGE(fc_queue_head) % QUEUE_CLEANUP_DELAY == 0)
				tryAdvanceTail = true;

			/* 并退出循环 */
			break;
		}
	}

	/* 成功，因此更新全局 QUEUE_HEAD */
	QUEUE_HEAD = fc_queue_head;

	LWLockRelease(NotifySLRULock);

	return fc_nextNotify;
}

/*
 * SQL 函数以返回当前占用的通知队列的比例。
 */
Datum pg_notification_queue_usage(PG_FUNCTION_ARGS)
{
	double		fc_usage;

	/* 推进队列尾，以便我们不报告一个过大的结果 */
	fc_asyncQueueAdvanceTail();

	LWLockAcquire(NotifyQueueLock, LW_SHARED);
	fc_usage = fc_asyncQueueUsage();
	LWLockRelease(NotifyQueueLock);

	PG_RETURN_FLOAT8(fc_usage);
}

/*
 * 返回当前占用的队列的比例。
 *
 * 调用者必须以（至少）共享模式持有 NotifyQueueLock。
 *
 * 注意：我们测量距离逻辑尾页面的距离，而不是物理
 * 尾页面。从某种意义上说，这是错误的，但物理尾的位置受到 SLRU 段边界等细节的影响，
 * 因此基于此的结果是令人不快的不稳定。
 */
static double fc_asyncQueueUsage(void)
{
	int			fc_headPage = QUEUE_POS_PAGE(QUEUE_HEAD);
	int			fc_tailPage = QUEUE_POS_PAGE(QUEUE_TAIL);
	int			fc_occupied;

	fc_occupied = fc_headPage - fc_tailPage;

	if (fc_occupied == 0)
		return (double) 0;		/* 快速退出常见情况 */

	if (fc_occupied < 0)
	{
		/* 头部已经包裹，尾部尚未 */
		fc_occupied += QUEUE_MAX_PAGE + 1;
	}

	return (double) fc_occupied / (double) ((QUEUE_MAX_PAGE + 1) / 2);
}

/*
 * 检查队列是否至少满了一半，如果是的话发出警告。
 *
 * 鉴于队列的大小，这种情况不太可能，但仍然可能。
 * 警告最多每 QUEUE_FULL_WARN_INTERVAL 显示一次。
 *
 * 调用者必须保持独占的 NotifyQueueLock。
 */
static void fc_asyncQueueFillWarning(void)
{
	double		fc_fillDegree;
	TimestampTz fc_t;

	fc_fillDegree = fc_asyncQueueUsage();
	if (fc_fillDegree < 0.5)
		return;

	fc_t = GetCurrentTimestamp();

	if (TimestampDifferenceExceeds(asyncQueueControl->lastQueueFillWarn,
								   fc_t, QUEUE_FULL_WARN_INTERVAL))
	{
		QueuePosition fc_min = QUEUE_HEAD;
		int32		fc_minPid = InvalidPid;

		for (BackendId fc_i = QUEUE_FIRST_LISTENER; fc_i > 0; fc_i = QUEUE_NEXT_LISTENER(fc_i))
		{
			Assert(QUEUE_BACKEND_PID(fc_i) != InvalidPid);
			fc_min = QUEUE_POS_MIN(fc_min, QUEUE_BACKEND_POS(fc_i));
			if (QUEUE_POS_EQUAL(fc_min, QUEUE_BACKEND_POS(fc_i)))
				fc_minPid = QUEUE_BACKEND_PID(fc_i);
		}

		ereport(WARNING,
				(errmsg("NOTIFY queue is %.0f%% full", fc_fillDegree * 100),
				 (fc_minPid != InvalidPid ?
				  errdetail("The server process with PID %d is among those with the oldest transactions.", fc_minPid)
				  : 0),
				 (fc_minPid != InvalidPid ?
				  errhint("The NOTIFY queue cannot be emptied until that process ends its current transaction.")
				  : 0)));

		asyncQueueControl->lastQueueFillWarn = fc_t;
	}
}

/*
 * 向侦听的后端发送信号。
 *
 * 通常我们只向我们自己数据库中的后端发送信号，因为只有这些
 * 后端可能对我们发送的通知感兴趣。然而，如果在我们的数据库中有
 * 通知流量但在另一个有侦听者的数据库中没有流量，这些侦听者将会
 * 越来越落后。如果他们落后得足够多，还是要唤醒他们，以便他们
 * 可以推进自己的队列位置指针，从而允许全局尾部推进。
 *
 * 由于我们知道 BackendId 和 Pid，信号发送的成本相当低。
 *
 * 这个操作在 CommitTransaction() 期间调用，因此它
 * 失败的概率非常低是很重要的。
 */
static void fc_SignalBackends(void)
{
	int32	   *fc_pids;
	BackendId  *fc_ids;
	int			fc_count;

	/*
	 * 识别我们需要发送信号的后端。我们不想在持有 NotifyQueueLock
	 * 时发送信号，因此这个循环仅仅构建一个目标 PIDs 的列表。
	 *
	 * XXX 原则上这些 pallocs 可能会失败，这将是坏事。也许
	 * 预分配这些数组？不过它们不是那么大。
	 */
	fc_pids = (int32 *) palloc(MaxBackends * sizeof(int32));
	fc_ids = (BackendId *) palloc(MaxBackends * sizeof(BackendId));
	fc_count = 0;

	LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
	for (BackendId fc_i = QUEUE_FIRST_LISTENER; fc_i > 0; fc_i = QUEUE_NEXT_LISTENER(fc_i))
	{
		int32		fc_pid = QUEUE_BACKEND_PID(fc_i);
		QueuePosition fc_pos;

		Assert(fc_pid != InvalidPid);
		fc_pos = QUEUE_BACKEND_POS(fc_i);
		if (QUEUE_BACKEND_DBOID(fc_i) == MyDatabaseId)
		{
			/*
			 * 除非监听者已经赶上（不太可能，但可能），否则始终
			 * 向我们自己数据库中的监听者发送信号。
			 */
			if (QUEUE_POS_EQUAL(fc_pos, QUEUE_HEAD))
				continue;
		}
		else
		{
			/*
			 * 只有当其他数据库中的监听者落后得很远时，
			 * 才应该向他们发送信号。
			 */
			if (fc_asyncQueuePageDiff(QUEUE_POS_PAGE(QUEUE_HEAD),
								   QUEUE_POS_PAGE(fc_pos)) < QUEUE_CLEANUP_DELAY)
				continue;
		}
		/* 好的，需要给这个发送信号 */
		fc_pids[fc_count] = fc_pid;
		fc_ids[fc_count] = fc_i;
		fc_count++;
	}
	LWLockRelease(NotifyQueueLock);

	/* 现在发送信号 */
	for (int fc_i = 0; fc_i < fc_count; fc_i++)
	{
		int32		fc_pid = fc_pids[fc_i];

		/*
		 * 如果我们正在向自己的进程发送信号，则无需涉及内核；
		 * 直接设置标志即可。
		 */
		if (fc_pid == MyProcPid)
		{
			notifyInterruptPending = true;
			continue;
		}

		/*
		 * 注意：假设事情没有损坏，信号在这里的失败只可能发生在
		 * 目标后端在我们释放 NotifyQueueLock 后退出；这不太可能但
		 * 绝对可能。所以如果发生这种情况，我们只需记录一个低级别
		 * 的调试消息。
		 */
		if (SendProcSignal(fc_pid, PROCSIG_NOTIFY_INTERRUPT, fc_ids[fc_i]) < 0)
			elog(DEBUG3, "could not signal backend with PID %d: %m", fc_pid);
	}

	pfree(fc_pids);
	pfree(fc_ids);
}

/*
 * AtAbort_Notify
 *
 *	在事务中止时调用此函数。
 *
 *	消除我们在事务提交时会执行的挂起操作和出站通知。
 */
void AtAbort_Notify(void)
{
	/*
	 * 如果我们监听但在 PreCommit_Notify 之后回滚事务，
	 * 我们已经注册为监听者但没有在 listenChannels 中
	 * 进行任何条目。在这种情况下，重新注销。
	 */
	if (amRegisteredListener && listenChannels == NIL)
		fc_asyncQueueUnregister();

	/* 并进行清理 */
	fc_ClearPendingActionsAndNotifies();
}

/*
 * AtSubCommit_Notify() --- 处理子事务提交。
 *
 * 将待处理列表中的所有项重新分配给父事务。
 */
void AtSubCommit_Notify(void)
{
	int			fc_my_level = GetCurrentTransactionNestLevel();

	/* 如果在我们的嵌套级别有操作，我们必须重新父化它们。 */
	if (pendingActions != NULL &&
		pendingActions->nestingLevel >= fc_my_level)
	{
		if (pendingActions->upper == NULL ||
			pendingActions->upper->nestingLevel < fc_my_level - 1)
		{
			/* 没有什么可以合并的；把整个东西给父级 */
			--pendingActions->nestingLevel;
		}
		else
		{
			ActionList *fc_childPendingActions = pendingActions;

			pendingActions = pendingActions->upper;

			/*
			 * 这里不得尝试消除重复项 --- 请参见 queue_listen()
			 */
			pendingActions->actions =
				list_concat(pendingActions->actions,
							fc_childPendingActions->actions);
			pfree(fc_childPendingActions);
		}
	}

	/* 如果在我们的嵌套级别有通知，我们必须重新父化它们。 */
	if (pendingNotifies != NULL &&
		pendingNotifies->nestingLevel >= fc_my_level)
	{
		Assert(pendingNotifies->nestingLevel == fc_my_level);

		if (pendingNotifies->upper == NULL ||
			pendingNotifies->upper->nestingLevel < fc_my_level - 1)
		{
			/* 没有什么可以合并的；把整个东西给父级 */
			--pendingNotifies->nestingLevel;
		}
		else
		{
			/*
			 * 以前，我们不想在这里消除重复项，但
			 * 现在我们必须这样做，否则我们将违反“Assert(!found)”，可能在这里
			 * 或在稍后的构建父级哈希表的尝试中。
			 */
			NotificationList *fc_childPendingNotifies = pendingNotifies;
			ListCell   *fc_l;

			pendingNotifies = pendingNotifies->upper;
			/* Insert all the subxact's events into parent, except for dups */
			foreach(fc_l, fc_childPendingNotifies->events)
			{
				Notification *fc_childn = (Notification *) lfirst(fc_l);

				if (!fc_AsyncExistsPendingNotify(fc_childn))
					fc_AddEventToPendingNotifies(fc_childn);
			}
			pfree(fc_childPendingNotifies);
		}
	}
}

/*
 * AtSubAbort_Notify() --- 处理子事务中止。
 */
void AtSubAbort_Notify(void)
{
	int			fc_my_level = GetCurrentTransactionNestLevel();

	/*
	 * 我们要做的就是弹出堆栈 --- 在此子事务中所做的操作/通知
	 * 不再重要，当CurTransactionContext被回收时空间将被释放。
	 * 不过，我们仍然必须释放
	 * ActionList和NotificationList对象，因为
	 * 这些是在TopTransactionContext中分配的。
	 *
	 * 请注意，可能根本没有条目，或者对于
	 * 当前子事务级别没有条目，可能是因为从未创建过，或者
	 * 因为我们在子事务中止期间因故重新进入此例程。
	 */
	while (pendingActions != NULL &&
		   pendingActions->nestingLevel >= fc_my_level)
	{
		ActionList *fc_childPendingActions = pendingActions;

		pendingActions = pendingActions->upper;
		pfree(fc_childPendingActions);
	}

	while (pendingNotifies != NULL &&
		   pendingNotifies->nestingLevel >= fc_my_level)
	{
		NotificationList *fc_childPendingNotifies = pendingNotifies;

		pendingNotifies = pendingNotifies->upper;
		pfree(fc_childPendingNotifies);
	}
}

/*
 * HandleNotifyInterrupt
 *
 *		信号处理部分的中断处理。让后端知道
 *		有待处理的通知中断。如果我们当前正在从客户端读取，
 *		这将中断读取，并且
 *		ProcessClientReadInterrupt()将调用ProcessNotifyInterrupt()。
 */
void HandleNotifyInterrupt(void)
{
	/*
	 * 注意：这是由信号处理程序调用的。您必须非常小心在这里做什么。
	 */

	/* signal that work needs to be done */
	notifyInterruptPending = true;

	/* make sure the event is processed in due course */
	SetLatch(MyLatch);
}

/*
 * ProcessNotifyInterrupt
 *
 *		如果我们看到notifyInterruptPending设置，将在
 *		在前端命令结束时传输ReadyForQuery之前调用此函数，
 *		如果在从前端读取时发生通知信号也会调用。
 *		HandleNotifyInterrupt()将通过进程的闩锁中断读取，
 *		并将调用此例程。
 *		如果我们真的空闲（即不是在事务块内），
 *		处理传入的通知。
 *
 *		如果“flush”为真，强制立即发送任何前端消息。
 *		在前端命令结束时调用时，这可能为假，
 *		因为我们在发送ReadyForQuery后将刷新。
 */
void ProcessNotifyInterrupt(bool fc_flush)
{
	if (IsTransactionOrTransactionBlock())
		return;					/* not really idle */

	/* Loop in case another signal arrives while sending messages */
	while (notifyInterruptPending)
		fc_ProcessIncomingNotify(fc_flush);
}


/*
 * 从队列中读取所有待处理的通知，并将适当的
 * 通知发送到我的前端。当我们到达队列头或未提交的
 * 通知时停止。
 */
static void fc_asyncQueueReadAllNotifications(void)
{
	volatile QueuePosition fc_pos;
	QueuePosition fc_head;
	Snapshot	fc_snapshot;

	/* page_buffer must be adequately aligned, so use a union */
	union
	{
		char		buf[QUEUE_PAGESIZE];
		AsyncQueueEntry align;
	}			fc_page_buffer;

	/* Fetch current state */
	LWLockAcquire(NotifyQueueLock, LW_SHARED);
	/* Assert checks that we have a valid state entry */
	Assert(MyProcPid == QUEUE_BACKEND_PID(MyBackendId));
	fc_pos = QUEUE_BACKEND_POS(MyBackendId);
	fc_head = QUEUE_HEAD;
	LWLockRelease(NotifyQueueLock);

	if (QUEUE_POS_EQUAL(fc_pos, fc_head))
	{
		/* Nothing to do, we have read all notifications already. */
		return;
	}

	/*----------
	 * 获取快照，我们将用来决定哪些事务仍在进行中。
	 * 这比看起来要复杂，因为存在竞争条件。
	 * 考虑以下示例：
	 *
	 * 后端 1:					 后端 2:
	 *
	 * 事务开始
	 * UPDATE foo SET ...;
	 * NOTIFY foo;
	 * 提交开始
	 * 排队通知消息
	 *								 事务开始
	 *								 LISTEN foo;  -- 会话中的第一次 LISTEN
	 *								 SELECT * FROM foo WHERE ...;
	 * 提交到 clog
	 *								 提交开始
	 *								 将后端 2 添加到监听器数组
	 *								 移动到队列头部（这段代码）
	 *								 提交到 clog
	 *
	 * 事务 2 的 SELECT 尚未看到 UPDATE 的效果，因为它尚未提交。
	 * 理想情况下，我们会确保客户端 2 最终会收到事务 1 的通知消息，但这没有办法做到；
	 * 在我们进入监听器数组之前，无法保证通知消息不会从队列中移除。
	 *
	 * 因此，事务 2 正在使用的编码技术是不安全的：
	 * 应用程序必须在检查数据库状态之前提交 LISTEN，
	 * 如果他们想确保看到关于随后的状态变化的通知。
	 *
	 * 我们保证的是，我们将看到在这里拍摄的快照之后提交的
	 * 事务的所有通知。
	 * Exec_ListenPreCommit 已经将我们添加到监听器数组中，
	 * 因此在我们看到它们之前，没有尚未提交的消息可以从队列中移除。
	 *----------
	 */
	fc_snapshot = RegisterSnapshot(GetLatestSnapshot());

	/*
	 * 在尝试将消息发送到我们的前端时，我们可能会失败
	 * （例如，由于编码转换失败）。如果发生这种情况，
	 * 我们不能一遍又一遍地尝试发送相同的消息，这一点很重要。
	 * 因此，我们在这里放置一个 PG_TRY 块，该块将在
	 * 我们失去对错误的控制之前强制前进我们的队列位置。
	 * （我们也可以在处理每个单独的消息之前重新获取 NotifyQueueLock
	 * 并移动位置，但那样似乎锁的交通太多。）
	 */
	PG_TRY();
	{
		bool		fc_reachedStop;

		do
		{
			int			fc_curpage = QUEUE_POS_PAGE(fc_pos);
			int			fc_curoffset = QUEUE_POS_OFFSET(fc_pos);
			int			fc_slotno;
			int			fc_copysize;

			/*
			 * 我们将数据从 SLRU 复制到本地缓冲区，以避免在检查
			 * 条目并可能将其传输到我们的前端时保持 NotifySLRULock。
			 * 只复制我们将实际检查的页面部分。
			 */
			fc_slotno = SimpleLruReadPage_ReadOnly(NotifyCtl, fc_curpage,
												InvalidTransactionId);
			if (fc_curpage == QUEUE_POS_PAGE(fc_head))
			{
				/* 我们只想读取到头部 */
				fc_copysize = QUEUE_POS_OFFSET(fc_head) - fc_curoffset;
				if (fc_copysize < 0)
					fc_copysize = 0;	/* 只是为了安全 */
			}
			else
			{
				/* 获取页面的其余部分 */
				fc_copysize = QUEUE_PAGESIZE - fc_curoffset;
			}
			memcpy(fc_page_buffer.buf + fc_curoffset,
				   NotifyCtl->shared->page_buffer[fc_slotno] + fc_curoffset,
				   fc_copysize);
			/* 释放我们从 SimpleLruReadPage_ReadOnly() 获取的锁 */
			LWLockRelease(NotifySLRULock);

			/*
			 * 处理消息直到停止位置、页面末尾或未提交的消息。
			 *
			 * 我们的停止位置是我们进入此函数时找到的头的位置。
			 * 它可能已经改变了。
			 * 但如果改变了，我们将收到（或已经收到并排队）另一个信号，重新来到这里。
			 *
			 * 我们在这里没有持有 NotifyQueueLock！队列只能延伸到头指针之外（见上文），
			 * 我们保持后端的指针不变，以便没有人会在我们之下截断或重写页面。
			 * 特别是我们不想在向前端发送通知时持有锁。
			 */
			fc_reachedStop = fc_asyncQueueProcessPageEntries(&fc_pos, fc_head,
													   fc_page_buffer.buf,
													   fc_snapshot);
		} while (!fc_reachedStop);
	}
	PG_FINALLY();
	{
		/* 更新共享状态 */
		LWLockAcquire(NotifyQueueLock, LW_SHARED);
		QUEUE_BACKEND_POS(MyBackendId) = fc_pos;
		LWLockRelease(NotifyQueueLock);
	}
	PG_END_TRY();

	/* 快照处理完成 */
	UnregisterSnapshot(fc_snapshot);
}

/*
 * 从共享队列中获取通知，从当前位置开始，并将相关的通知传递给我的前端。
 *
 * 当前页面必须已经从共享内存中提取到 page_buffer 中。
 * （我们可以直接在共享内存中访问页面，但那样会意味着在此例程中持有 NotifySLRULock。）
 *
 * 如果达到“停止”位置，或遇到未提交事务的通知，或到达页面末尾，我们将停止。
 *
 * 一旦我们到达停止位置或未提交通知，该函数返回 true，
 * 如果我们已经完成页面处理则返回 false。
 * 换句话说：一旦返回 true，就没有必要进一步查看。
 * QueuePosition *current 会在所有已处理的消息上推进。
 */
static bool fc_asyncQueueProcessPageEntries(volatile QueuePosition *fc_current,
							 QueuePosition fc_stop,
							 char *fc_page_buffer,
							 Snapshot fc_snapshot)
{
	bool		fc_reachedStop = false;
	bool		fc_reachedEndOfPage;
	AsyncQueueEntry *fc_qe;

	do
	{
		QueuePosition fc_thisentry = *fc_current;

		if (QUEUE_POS_EQUAL(fc_thisentry, fc_stop))
			break;

		fc_qe = (AsyncQueueEntry *) (fc_page_buffer + QUEUE_POS_OFFSET(fc_thisentry));

		/*
		 * 将 *current 前移至该消息，可能到达下一页面。如
		 * asyncQueueReadAllNotifications 的注释所述，我们必须在可能在处理消息时失败之前执行此操作。
		 */
		fc_reachedEndOfPage = fc_asyncQueueAdvance(fc_current, fc_qe->length);

		/* 忽略发往其他数据库的消息 */
		if (fc_qe->dboid == MyDatabaseId)
		{
			if (XidInMVCCSnapshot(fc_qe->xid, fc_snapshot))
			{
				/*
				 * 源事务仍在进行中，因此我们还无法处理此消息。
				 * 退出循环，但首先将 *current 向后移动，以便下次重新处理该消息。
				 * （注意：虽然不太可能，但 TransactionIdDidCommit 失败并非不可能，
				 * 因此在处理未提交消息时，我们无法避免此前推进然后回退的行为。）
				 *
				 * 注意，我们必须先测试 XidInMVCCSnapshot，
				 * 然后再测试 TransactionIdDidCommit，否则我们可能会返回一个尚未对快照可见的
				 * 事务的消息；请参阅 heapam_visibility.c 开头的注释。
				 *
				 * 此外，虽然我们自己的事务不会在快照中列出，
				 * 但我们无需检查 TransactionIdIsCurrentTransactionId，
				 * 因为我们的事务尚未排队任何消息。
				 */
				*fc_current = fc_thisentry;
				fc_reachedStop = true;
				break;
			}
			else if (TransactionIdDidCommit(fc_qe->xid))
			{
				/* qe->data 是以空字符终止的频道名称 */
				char	   *fc_channel = fc_qe->data;

				if (fc_IsListeningOn(fc_channel))
				{
					/* 负载在频道名称之后 */
					char	   *fc_payload = fc_qe->data + strlen(fc_channel) + 1;

					NotifyMyFrontEnd(fc_channel, fc_payload, fc_qe->srcPid);
				}
			}
			else
			{
				/*
				 * 源事务已中止或崩溃，所以我们只需
				 * 忽略其通知。
				 */
			}
		}

		/* 如果我们还未到达页面末尾，则循环返回 */
	} while (!fc_reachedEndOfPage);

	if (QUEUE_POS_EQUAL(*fc_current, fc_stop))
		fc_reachedStop = true;

	return fc_reachedStop;
}

/*
 * 将共享队列尾变量推进到所有每个后端尾指针的最小值。如果可能，截断 pg_notify 空间。
 *
 * 这个函数（通常）在 CommitTransaction() 中调用，因此它的失败概率必须非常低。
 */
static void fc_asyncQueueAdvanceTail(void)
{
	QueuePosition fc_min;
	int			fc_oldtailpage;
	int			fc_newtailpage;
	int			fc_boundary;

	/* 限制任务在每个集群中只有一个后端；参见 SimpleLruTruncate()。 */
	LWLockAcquire(NotifyQueueTailLock, LW_EXCLUSIVE);

	/*
	 * 计算新的尾部。在 v13 之前，QUEUE_TAIL 必须是准确的
	 * （即，至少与一个后端的队列位置完全匹配），因此它必须
	 * 与实际计算原子更新。自 v13 起，我们可以不这样做，但保持
	 * 这样似乎更谨慎。
	 *
	 * 此外，由于传入的后端将从 QUEUE_TAIL 向前扫描，因此
	 * 必须在我们可以截断任何数据之前推进 QUEUE_TAIL。因此，QUEUE_TAIL 是
	 * 逻辑尾部，而 QUEUE_STOP_PAGE 是物理尾部，即最旧的
	 * 未截断页面。当 QUEUE_STOP_PAGE != QUEUE_POS_PAGE(QUEUE_TAIL) 时，
	 * 还有页面可以截断但尚未完成截断。
	 *
	 * 出于并发考虑，我们不想在执行 SimpleLruTruncate 时持有 NotifyQueueLock。
	 * 这样是可以的，因为没有后端会尝试访问我们正在截断的页面。
	 */
	LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
	fc_min = QUEUE_HEAD;
	for (BackendId fc_i = QUEUE_FIRST_LISTENER; fc_i > 0; fc_i = QUEUE_NEXT_LISTENER(fc_i))
	{
		Assert(QUEUE_BACKEND_PID(fc_i) != InvalidPid);
		fc_min = QUEUE_POS_MIN(fc_min, QUEUE_BACKEND_POS(fc_i));
	}
	QUEUE_TAIL = fc_min;
	fc_oldtailpage = QUEUE_STOP_PAGE;
	LWLockRelease(NotifyQueueLock);

	/*
	 * 如果全局尾部跨越了 SLRU 段边界，我们可以截断某些内容。
	 *
	 * XXX 为了减少目录扫描的次数，可能每几个段截断一次更好。
	 */
	fc_newtailpage = QUEUE_POS_PAGE(fc_min);
	fc_boundary = fc_newtailpage - (fc_newtailpage % SLRU_PAGES_PER_SEGMENT);
	if (fc_asyncQueuePagePrecedes(fc_oldtailpage, fc_boundary))
	{
		/*
		 * SimpleLruTruncate() 将请求 NotifySLRULock，但也将
		 * 再次释放该锁。
		 */
		SimpleLruTruncate(NotifyCtl, fc_newtailpage);

		/*
		 * 更新 QUEUE_STOP_PAGE。这会改变 asyncQueueIsFull() 对于
		 * 旧尾部之前段的裁定，允许新数据进入该段。
		 */
		LWLockAcquire(NotifyQueueLock, LW_EXCLUSIVE);
		QUEUE_STOP_PAGE = fc_newtailpage;
		LWLockRelease(NotifyQueueLock);
	}

	LWLockRelease(NotifyQueueTailLock);
}

/*
 * ProcessIncomingNotify
 *
 *		扫描队列以查找到达的通知并将其报告给前端。这些通知可能来自其他会话，
 *		或我们自己；这里没有必要区分。
 *
 *		如果 "flush" 为真，立即强制将任何前端消息送出。
 *
 *		注意：由于我们处于任何事务之外，必须创建自己的事务。
 */
static void fc_ProcessIncomingNotify(bool fc_flush)
{
	MemoryContext fc_oldcontext;

	/* 我们 *必须* 重置标志 */
	notifyInterruptPending = false;

	/* 如果我们没有积极监听，不做其他事情 */
	if (listenChannels == NIL)
		return;

	if (Trace_notify)
		elog(DEBUG1, "ProcessIncomingNotify");

	set_ps_display("notify interrupt");

	/*
	 * 我们必须在事务中运行 asyncQueueReadAllNotifications，否则
	 * 如果出现错误，将会发生不好的事情。然而，我们需要保留
	 * 调用者的内存上下文（通常是 MessageContext）。
	 */
	fc_oldcontext = CurrentMemoryContext;

	StartTransactionCommand();

	fc_asyncQueueReadAllNotifications();

	CommitTransactionCommand();

	/* 调用者的上下文最好不要是事务本地的 */
	Assert(MemoryContextIsValid(fc_oldcontext));
	MemoryContextSwitchTo(fc_oldcontext);

	/*
	 * 如果这不是一个命令结束情况，必须刷新通知消息
	 * 以确保前端及时获取它们。
	 */
	if (fc_flush)
		pq_flush();

	set_ps_display("idle");

	if (Trace_notify)
		elog(DEBUG1, "ProcessIncomingNotify: done");
}

/*
 * 向我的前端发送 NOTIFY 消息。
 */
void NotifyMyFrontEnd(const char *fc_channel, const char *fc_payload, int32 fc_srcPid)
{
	if (whereToSendOutput == DestRemote)
	{
		StringInfoData fc_buf;

		pq_beginmessage(&fc_buf, 'A');
		pq_sendint32(&fc_buf, fc_srcPid);
		pq_sendstring(&fc_buf, fc_channel);
		pq_sendstring(&fc_buf, fc_payload);
		pq_endmessage(&fc_buf);

		/*
		 * 注意：我们在这里不执行 pq_flush()。某个层级的调用者将在
		 * 之后处理它，使得此消息可以与其他消息合并成一个数据包。
		 */
	}
	else
		elog(INFO, "NOTIFY for \"%s\" payload \"%s\"", fc_channel, fc_payload);
}

/* pendingNotifies 是否包含给定事件的匹配项？ */
static bool fc_AsyncExistsPendingNotify(Notification *fc_n)
{
	if (pendingNotifies == NULL)
		return false;

	if (pendingNotifies->hashtab != NULL)
	{
		/* 使用哈希表来查找匹配项 */
		if (hash_search(pendingNotifies->hashtab,
						&fc_n,
						HASH_FIND,
						NULL))
			return true;
	}
	else
	{
		/* 必须扫描事件列表 */
		ListCell   *fc_l;

		foreach(fc_l, pendingNotifies->events)
		{
			Notification *fc_oldn = (Notification *) lfirst(fc_l);

			if (fc_n->channel_len == fc_oldn->channel_len &&
				fc_n->payload_len == fc_oldn->payload_len &&
				memcmp(fc_n->data, fc_oldn->data,
					   fc_n->channel_len + fc_n->payload_len + 2) == 0)
				return true;
		}
	}

	return false;
}

/*
 * 将通知事件添加到预先存在的 pendingNotifies 列表中。
 *
 * 因为 pendingNotifies->events 已经非空，这样做
 * 无论 CurrentMemoryContext 是什么，都是正确的。
 */
static void fc_AddEventToPendingNotifies(Notification *fc_n)
{
	Assert(pendingNotifies->events != NIL);

	/* 如果需要，创建哈希表 */
	if (list_length(pendingNotifies->events) >= MIN_HASHABLE_NOTIFIES &&
		pendingNotifies->hashtab == NULL)
	{
		HASHCTL		fc_hash_ctl;
		ListCell   *fc_l;

		/* 创建哈希表 */
		fc_hash_ctl.keysize = sizeof(Notification *);
		fc_hash_ctl.entrysize = sizeof(NotificationHash);
		fc_hash_ctl.hash = fc_notification_hash;
		fc_hash_ctl.match = fc_notification_match;
		fc_hash_ctl.hcxt = CurTransactionContext;
		pendingNotifies->hashtab =
			hash_create("Pending Notifies",
						256L,
						&fc_hash_ctl,
						HASH_ELEM | HASH_FUNCTION | HASH_COMPARE | HASH_CONTEXT);

		/* 插入所有已经存在的事件 */
		foreach(fc_l, pendingNotifies->events)
		{
			Notification *fc_oldn = (Notification *) lfirst(fc_l);
			NotificationHash *fc_hentry;
			bool		fc_found;

			fc_hentry = (NotificationHash *) hash_search(pendingNotifies->hashtab,
													  &fc_oldn,
													  HASH_ENTER,
													  &fc_found);
			Assert(!fc_found);
			fc_hentry->event = fc_oldn;
		}
	}

	/* 按顺序将新事件添加到列表中 */
	pendingNotifies->events = lappend(pendingNotifies->events, fc_n);

	/* 如果需要，将事件添加到哈希表 */
	if (pendingNotifies->hashtab != NULL)
	{
		NotificationHash *fc_hentry;
		bool		fc_found;

		fc_hentry = (NotificationHash *) hash_search(pendingNotifies->hashtab,
												  &fc_n,
												  HASH_ENTER,
												  &fc_found);
		Assert(!fc_found);
		fc_hentry->event = fc_n;
	}
}

/*
 * notification_hash：用于通知哈希表的哈希函数
 *
 * 哈希“键”是指向 Notification 结构的指针。
 */
static uint32 fc_notification_hash(const void *fc_key, Size fc_keysize)
{
	const Notification *fc_k = *(const Notification *const *) fc_key;

	Assert(fc_keysize == sizeof(Notification *));
	/* 我们不在哈希中包含有效负载的尾部空字符 */
	return DatumGetUInt32(hash_any((const unsigned char *) fc_k->data,
								   fc_k->channel_len + fc_k->payload_len + 1));
}

/*
 * notification_match：用于 notification_hash 的匹配函数
 */
static int fc_notification_match(const void *fc_key1, const void *fc_key2, Size fc_keysize)
{
	const Notification *fc_k1 = *(const Notification *const *) fc_key1;
	const Notification *fc_k2 = *(const Notification *const *) fc_key2;

	Assert(fc_keysize == sizeof(Notification *));
	if (fc_k1->channel_len == fc_k2->channel_len &&
		fc_k1->payload_len == fc_k2->payload_len &&
		memcmp(fc_k1->data, fc_k2->data,
			   fc_k1->channel_len + fc_k1->payload_len + 2) == 0)
		return 0;				/* 等于 */
	return 1;					/* 不相等 */
}

/* 清除 pendingActions 和 pendingNotifies 列表。 */
static void fc_ClearPendingActionsAndNotifies(void)
{
	/*
	 * 所有内容都分配在 TopTransactionContext 或与之对应的子事务的上下文中。
	 * 所以，这里除了重置指针之外没有其他要做的；
	 * 空间将在上下文被删除时被回收。
	 */
	pendingActions = NULL;
	pendingNotifies = NULL;
}
